package psnl.bingo.mr.demo4;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.time.LocalDateTime;

public class Demo4App extends Configured implements Tool {
    public static void main(String[] args) {
        Demo4App app = new Demo4App();

        Configuration conf = new Configuration();

        try {
            Path outPath = new Path(args[1]);
            FileSystem fs = FileSystem.get(conf);

            if (fs.exists(outPath)) {
                System.out.println("output dir exists, delete...");
                fs.delete(outPath, true);
                System.out.println("output dir deleted success");
            }

            System.out.println(LocalDateTime.now() + "--start running...");
            int status = ToolRunner.run(conf, app, args);
            System.out.println(LocalDateTime.now() + "--task run over. success = " + (status==0));
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(this.getConf(), "Demo4App");
        job.setJarByClass(this.getClass());

        // --------- 配置 MRApp ----------------
        Path input = new Path(args[0]);
        Path output = new Path(args[1]);

        Class mapperClz = D4Mapper.class; // mapper Class
        Class reducerClz = D4Reducer.class; // reducer Class
        Class mapOutKeyClz = D4Line.class; // mapper输出的key Class
        Class mapOutValueClz = NullWritable.class; // mapper输出的value Class
        Class outKeyClz = D4Line.class; // reducer输出的key Class
        Class outValueClz = Text.class; // reducer输出的value Class
        Class groupComparator = D4MapOutGroup.class; // mapper输出数据的分组器


        // --------- 配置 MRApp ----------------

        job.setMapperClass(mapperClz);
        job.setReducerClass(reducerClz);

        job.setMapOutputKeyClass(mapOutKeyClz);
        job.setMapOutputValueClass(mapOutValueClz);

        job.setOutputKeyClass(outKeyClz);
        job.setOutputValueClass(outValueClz);

        job.setGroupingComparatorClass(groupComparator);

        // 设置自定义的数据输入格式处理类
        job.setInputFormatClass(D4InputFormat.class);
        // 设置自定义的数据输出格式处理类
        job.setOutputFormatClass(D4OutputFormat.class);
        D4RecordWriter.outPathDir = output.toString();

        FileInputFormat.setInputPaths(job, input);
        FileOutputFormat.setOutputPath(job, output);

        boolean success = job.waitForCompletion(true);

        return success?0:1;
    }
}
